package iscas;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Pipeline;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

@SuppressWarnings("unused")
public class KeyByStrategy {
    private static volatile int keyByType, keyByRatio;
    private static final String leastKeyScript, leastCountScript, weightScript;
    private static final ConcurrentHashMap<Integer, Integer>
            leastKeyAssignMap = new ConcurrentHashMap<>(), leastCountAssignMap = new ConcurrentHashMap<>(),
            rangeAssignMap = new ConcurrentHashMap<>(), weightAssignMap = new ConcurrentHashMap<>();
    private static final ThreadLocal<Random> localRandom = ThreadLocal.withInitial(Random::new);
    private static final ArrayList<ArrayList<Integer>> reportList = new ArrayList<>(10);
    private static final int parallelismRedis, maxParallelismRedis;
    private static final JedisPool jedisPool;

    static {
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxIdle(40);
        config.setMaxTotal(60);
        config.setMinIdle(20);
        jedisPool = new JedisPool(config, RedisConfig.addr, 6379);

        Jedis jedis = RedisConfig.newJedis();
        keyByType = Integer.parseInt(RedisConfig.getOrDefault(jedis, "keyByType", "0"));
        keyByRatio = Integer.parseInt(RedisConfig.getOrDefault(jedis, "keyByRatio", "0"));
        parallelismRedis = Integer.parseInt(RedisConfig.getOrDefault(jedis, "parallelism", "0"));
        maxParallelismRedis = Integer.parseInt(RedisConfig.getOrDefault(jedis, "maxParallelism", "0"));
        leastKeyScript = jedis.get("leastKeyScript");
        leastCountScript = jedis.get("leastCountScript");
        weightScript = jedis.get("weightScript");
        jedis.close();
        System.out.printf("[ISCAS][RedisConfig] type: %d, ratio: %d, leastKey: %s, leastCount: %s, weight: %s, parallelism: %d, maxParallelism: %d\n",
                keyByType, keyByRatio, leastKeyScript, leastCountScript, weightScript, parallelismRedis, maxParallelismRedis);
        RedisConfig.subscribe(KeyByStrategy::subscribe, "keyByChannel");
        new Thread(() -> KeyByStrategy.report(parallelismRedis, maxParallelismRedis)).start();
        for (int i = 0; i < 10; i++) {
            reportList.add(new ArrayList<>());
        }
    }

    private static Jedis getJedis() {
        return jedisPool.getResource();
    }

    private static void subscribe(String msg) {
        String[] sa = msg.split("@");
        switch (sa[0]) {
            case "type":
                keyByType = Integer.parseInt(sa[1]);
                Jedis jedis = jedisPool.getResource();
                jedis.incr("keyByAck");
                jedis.close();
                break;
            case "ratio":
                keyByRatio = Integer.parseInt(sa[1]);
                break;
            case "clear":
                leastKeyAssignMap.clear();
                leastCountAssignMap.clear();
                rangeAssignMap.clear();
                weightAssignMap.clear();
                break;
            default:
                break;
        }
    }

    private static Integer incr(Integer k, Integer v) {
        return v == null ? 1 : v + 1;
    }

    @SuppressWarnings("SameParameterValue")
    private static void report(int parallelism, int maxParallelism) {
        String[] strategyName = {"hash", "module", "leastKey", "leastCount", "range", "weight", "custom", "count"};
        ArrayList<HashMap<Integer, Integer>> reportData = new ArrayList<>();
        Jedis jedis = RedisConfig.newJedis();
        for (int i = 0; i < strategyName.length; i++) {
            reportData.add(new HashMap<>());
        }
        int total = 0, oldTotal = 0;
        System.out.println("[ISCAS][Report Thread] created.");
        while (true) {
            try {
                for (ArrayList<Integer> list : reportList) {
                    ArrayList<Integer> t = new ArrayList<>();
                    // noinspection SynchronizationOnLocalVariableOrMethodParameter
                    synchronized (list) {
                        if (!list.isEmpty()) {
                            t.addAll(list);
                            total += list.size();
                            list.clear();
                        }
                    }
                    for (int hash : t) {
                        reportData.get(0).compute(hash(hash, parallelism, maxParallelism), KeyByStrategy::incr);
                        reportData.get(1).compute(module(hash, parallelism), KeyByStrategy::incr);
                        reportData.get(2).compute(leastKey(hash, parallelism), KeyByStrategy::incr);
                        reportData.get(3).compute(leastCount(hash, parallelism), KeyByStrategy::incr);
                        reportData.get(4).compute(range(hash, parallelism), KeyByStrategy::incr);
                        reportData.get(5).compute(weight(hash, parallelism), KeyByStrategy::incr);
                        reportData.get(6).compute(custom(hash, parallelism), KeyByStrategy::incr);
                        reportData.get(7).compute(hash, KeyByStrategy::incr);
                    }
                }
                for (int i = 0; i < strategyName.length; i++) {
                    if (reportData.get(i).isEmpty()) {
                        continue;
                    }
                    HashMap<Integer, Integer> data = reportData.get(i);
                    String name = strategyName[i];
                    Pipeline pipe = jedis.pipelined();
                    for (Integer key : data.keySet()) {
                        String hashName = "keyByStat_" + name;
                        data.compute(key, (Integer k, Integer v) -> {
                            if (v != null && v > 0) {
                                pipe.zincrby(hashName, v, String.valueOf(k));
                            }
                            return 0;
                        });
                    }
                    pipe.sync();
                }
                if (total != oldTotal) {
                    System.out.printf("[ISCAS][report] total: %d\n", total);
                    oldTotal = total;
                } else {
                    // noinspection BusyWait
                    Thread.sleep(500);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                jedis.close();
            }
        }
    }

    public static void report(int keyHash) {
        int rand = localRandom.get().nextInt(100);
        if (rand < keyByRatio) {
            rand %= reportList.size();
            synchronized (reportList.get(rand)) {
                reportList.get(rand).add(keyHash);
            }
        }
    }

    public static int assignByType(int keyHash, int parallelism, int maxParallelism) {
        int r;
        switch (keyByType) {
            case 0:
            default:
                r = hash(keyHash, parallelism, maxParallelism);
                break;
            case 1:
                r = module(keyHash, parallelism);
                break;
            case 2:
                r = leastKey(keyHash, parallelism);
                break;
            case 3:
                r = leastCount(keyHash, parallelism);
                break;
            case 4:
                r = range(keyHash, parallelism);
                break;
            case 5:
                r = weight(keyHash, parallelism);
                break;
            case 6:
                r = custom(keyHash, parallelism);
                break;
        }
        return r;
    }

    static int bitMix(int in) {
        in ^= in >>> 16;
        in *= 0x85ebca6b;
        in ^= in >>> 13;
        in *= 0xc2b2ae35;
        in ^= in >>> 16;
        return in;
    }

    static int murmurHash(int code) {
        code *= 0xcc9e2d51;
        code = Integer.rotateLeft(code, 15);
        code *= 0x1b873593;

        code = Integer.rotateLeft(code, 13);
        code = code * 5 + 0xe6546b64;

        code ^= 4;
        code = bitMix(code);

        if (code >= 0) {
            return code;
        } else if (code != Integer.MIN_VALUE) {
            return -code;
        } else {
            return 0;
        }
    }

    public static int computeOperatorIndexForKeyGroup(int maxParallelism, int parallelism, int keyGroupId) {
        return keyGroupId * parallelism / maxParallelism;
    }

    public static int hash(int keyHash, int parallelism, int maxParallelism) {
        return computeOperatorIndexForKeyGroup(maxParallelism, parallelism,
                                               murmurHash(keyHash) % maxParallelism);
    }

    public static int module(int keyHash, int parallelism) {
        return keyHash % parallelism;
    }

    private static int redisComputeAndCache(ConcurrentHashMap<Integer, Integer> map, int keyHash,
                                            int parallelism, String script) {
        map.computeIfAbsent(keyHash, (Integer k) -> {
            if (script == null) {
                return 0;
            }
            Jedis jedis = getJedis();
            String key1 = String.valueOf(k), key2 = String.valueOf(parallelism);
            Long r = (Long) jedis.evalsha(script, 2, key1, key2);
            jedis.close();
            return r.intValue();
        });
        return map.get(keyHash);
    }

    public static int leastKey(int keyHash, int parallelism) {
        return redisComputeAndCache(leastKeyAssignMap, keyHash, parallelism, leastKeyScript);
    }

    public static int leastCount(int keyHash, int parallelism) {
        return redisComputeAndCache(leastCountAssignMap, keyHash, parallelism, leastCountScript);
    }

    public static int range(int keyHash, int ignoredParallelism) {
        if (PassedVariables.range == null) {
            return 0;
        }
        rangeAssignMap.computeIfAbsent(keyHash, (Integer k) -> {
            int i = 0;
            while (k >= PassedVariables.range[i]) {
                i++;
            }
            return i;
        });
        return rangeAssignMap.get(keyHash);
    }

    public static int weight(int keyHash, int ignoredParallelism) {
        return redisComputeAndCache(weightAssignMap, keyHash, localRandom.get().nextInt(100), weightScript);
    }

    public static int custom(int keyHash, int parallelism) {
        if (PassedVariables.customKeyAssign != null) {
            return PassedVariables.customKeyAssign.apply(keyHash, parallelism);
        } else {
            return 0;
        }
    }
}
